#include <iostream>
#include <vector>
#include <cstring>
#include <cstdlib>
#include <queue>
#include <mutex>
#include <condition_variable>
#include <netinet/in.h>
#include <sys/epoll.h>
#include <sys/socket.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <pthread.h>

#define LISTEN_BACKLOG 20
#define MAXEVENTS 10
#define MAX_THREADS 4       // 线程池的最大线程数

using namespace std;

// 线程池任务结构体,包含文件描述符和接收到的数据
struct Task {
    int fd;                 // 文件描述符
    string data;            // 接收到的数据
};

// 线程池类
class ThreadPool {
public:
    ThreadPool() : stop(false) {} //构造函数，初始化停止标志为false
    ~ThreadPool() {   //析构函数，用于清理线程池
        stop = true;
        cv.notify_all();
        for (pthread_t &t : threads) {
            pthread_join(t, nullptr);
        }
    }

    void AddTask(Task task) {   //添加任务到线程池
        unique_lock<mutex> lock(mtx);
        tasks.emplace(move(task));
        cv.notify_one();
    }

    void Start() {    //启动线程池中的所有线程
        for (int i = 0; i < MAX_THREADS; ++i) {
            pthread_t t;
            if (pthread_create(&t, nullptr, &thread_function, this) != 0) {
                cerr << "Error creating thread" << endl;
                exit(EXIT_FAILURE);
            }
            threads.push_back(t);
        }
    }

    static void *thread_function(void *arg) {  //线程函数，处理任务队列中的任务
        ThreadPool *pool = static_cast<ThreadPool *>(arg);
        while (true) {
            Task task;
         {
            unique_lock<mutex> lock(pool->mtx);
            pool->cv.wait(lock, [pool] { return pool->stop || !pool->tasks.empty(); });
            if (pool->stop && pool->tasks.empty()) return nullptr;
            task = move(pool->tasks.front());
            pool->tasks.pop();
        }
            // 处理任务
            cout << "Thread processing data: " << task.data << endl;
            // 这里可以添加更多的逻辑来处理任务
            //向客户端发送数据
            const string response = "Server received your message: " + task.data;
            send(task.fd, response.c_str(), response.size(), 0);
        }
        return nullptr;
    }

private:
    vector<pthread_t> threads;   //存储线程ID的向量
    queue<Task> tasks;   //任务队列
    mutex mtx;    //互斥锁，用于同步任务队列
    condition_variable cv;   //条件变量，用于线程同步
    bool stop;   //停止标志
};

// 初始化套接字的函数
int init_socket(const char *ip, const char *port) {
    int listen_fd;
    struct sockaddr_in server_addr;
    socklen_t len = sizeof(server_addr);

    listen_fd = socket(AF_INET, SOCK_STREAM, 0);  //创建套接字
    if (listen_fd < 0) {
        cerr << "Error creating socket" << endl;
        return -1;
    }

    memset(&server_addr, 0, len);   //清零地址结构体
    server_addr.sin_family = AF_INET;  //设置地址族
    server_addr.sin_port = htons(stoi(port));   //设置端口号
    server_addr.sin_addr.s_addr = inet_addr(ip);  //设置ip地址

    if (bind(listen_fd, (struct sockaddr *)&server_addr, len) < 0) //绑定套接字
    {
        cerr << "Error binding socket" << endl;
        return -1;
    }

    if (listen(listen_fd, LISTEN_BACKLOG) < 0) {   //开始监听
        cerr << "Error listening on socket" << endl;
        return -1;
    }

    cout << "服务器已开启..." << endl;
    return listen_fd;
}

int main(int argc, char const *argv[]) {
    if (argc != 3) {    //检查命令行参数
        cerr << "Usage: " << argv[0] << " ip port" << endl;
        return -1;
    }

    int listen_fd = init_socket(argv[1], argv[2]);  //初始化套接字
    if (listen_fd < 0) {
        cerr << "Failed to initialize socket" << endl;
        return -1;
    }

    int epfd = epoll_create1(0);   //创建epoll实列
    if (epfd == -1) {
        cerr << "Failed to create epoll instance" << endl;
        return -1;
    }

    struct epoll_event ev, readyEV[MAXEVENTS];   //epoll事件结构体
    ev.events = EPOLLIN;   //设置监听读事件
    ev.data.fd = listen_fd;  //设置监听的文件描述符
    epoll_ctl(epfd, EPOLL_CTL_ADD, listen_fd, &ev);  //将监听套接字添加到epoll

    ThreadPool pool;   //创建线程池实列
    pool.Start();    //启动线程池

    while (1) {    //主循环，等待和处理epoll事件
        int nfds = epoll_wait(epfd, readyEV, MAXEVENTS, -1);  //等待事件
        if (nfds < 0) {
            cerr << "Error in epoll_wait" << endl;
            return -1;
        }
 struct sockaddr_in client_addr;   //客户端地址结构体

        for (int i = 0; i < nfds; ++i) {   //遍历所有事件
            if (readyEV[i].data.fd == listen_fd) {  //如果是监听套接字， 接受连接
                socklen_t len = sizeof(client_addr);
                int connect_fd = accept(listen_fd, (struct sockaddr *)&client_addr, &len);
                if (connect_fd < 0) {
                    cerr << "Failed to accept connection" << endl;
                    continue;
                }

                cout << "Connection successful!" << endl;
                cout << "Client fd = " << connect_fd << endl;
                cout << "Client IP = " << inet_ntoa(client_addr.sin_addr) << endl;
                cout << "Client port = " << ntohs(client_addr.sin_port) << endl;

                ev.events = EPOLLIN;    //设置监听读事件
                ev.data.fd = connect_fd;  //设置新的文件描述符
             epoll_ctl(epfd, EPOLL_CTL_ADD, connect_fd, &ev); //将新的连接添加到epoll
            } else if (readyEV[i].events == EPOLLIN) {  //如果是读事件
                char buf[1024] = {0};    //接收缓冲区
                int n = recv(readyEV[i].data.fd, buf, sizeof(buf), 0);
                if (n < 0) {
                    cerr << "Error receiving data" << endl;
                    continue;
                } else if (n == 0) {   //如果客户端关闭连接
                  cout << "Client IP = " << inet_ntoa(client_addr.sin_addr) << endl;
                  cout << "Client disconnected" << endl;
             epoll_ctl(epfd, EPOLL_CTL_DEL, readyEV[i].data.fd, nullptr); //从epoll中删除
                    close(readyEV[i].data.fd);
                } else {
                   cout << "Client IP = " << inet_ntoa(client_addr.sin_addr) << endl;
                    cout << "Received " << n << " bytes: " << buf << endl;
                    if (strncmp(buf, "quit", 4) == 0) {
                        epoll_ctl(epfd, EPOLL_CTL_DEL, readyEV[i].data.fd, nullptr);
                        close(readyEV[i].data.fd);  //关闭文件描述符
                    } else {   //否则，将任务添加到线程池
                        Task task{readyEV[i].data.fd, string(buf, n)};
                        pool.AddTask(move(task));
                    }
                }
            }
        }
    }

    return 0;
}